package com.itbaizhan.orders.message;

import com.alibaba.fastjson.JSONObject;
import com.itbaizhan.orders.entity.TxLog;
import com.itbaizhan.orders.mapper.TxLogMapper;
import com.itbaizhan.orders.service.IOrderService;
import com.itbaizhan.orders.tx.TxMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

/**
 * 监听事务消息
 */
//rocketmq如何监听消息  用注解：@RocketMQTransactionListener 监听哪个组
//实现RocketMQLocalTransactionListener接口
@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "tx_order_group")
public class OrderTxMessageListener implements RocketMQLocalTransactionListener {

    @Autowired
    private IOrderService iOrderService;
    @Resource
    private TxLogMapper txLogMapper;

    //接收rocketmq发来的消息
    @Transactional(rollbackFor = Exception.class)
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        log.info("******* 订单微服执行本地事务 **********");
        try {
            //获取消息
            TxMessage txMessage = this.getTxMessage(message);
            //3,执行本地事务
            iOrderService.sumbitOrderAndSaveTxNo(txMessage);
            log.info("******订单微服提交事务****");
            //提交事务  commit 生产者通知rocketmq   消息可以消费，即发送给消费者
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            e.printStackTrace();
            log.info("订单微服回滚事务");
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }

    /**
     * 回查事务，通过回查再提交一次，那样消息从未确认状态转为确认状态，可以发送出去
     * 因为网络异常或者其他原因，导致rocketmq 消息状态 UNKNOWN
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        log.info("********回查订单事务*********");
        //获取消息
        TxMessage txMessage = this.getTxMessage(message);

        //2,根据事务id查询信息
        TxLog txLog = txLogMapper.selectById(txMessage.getTxNo());
        //判断事务是否存在
        if(txLog != null){
            return RocketMQLocalTransactionState.COMMIT;
        }
        //unknown 生者通知rocketmq继续查询消息状态
        return RocketMQLocalTransactionState.UNKNOWN;

    }


    private TxMessage getTxMessage(Message message){
        //1,message 转 字符
        String messageStr = new String((byte[]) message.getPayload());
        //2,字符串转对象
        JSONObject jsonObject = JSONObject.parseObject(messageStr);
        //在发送事务时放进去的是自定义的message，所以现在从这key取出value
        String message1 = jsonObject.getString("message");
        TxMessage txMessage = JSONObject.parseObject(message1, TxMessage.class);
        return txMessage;
    }
}
